package cn._51doit.live.sources;

import cn._51doit.live.pojo.GiftBean;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;

/**
 * 自定义MySQL的Source存在问题
 * 1.该Source会不停的查询MySQL，如果每次查询的事件间隔太长，有延迟；太短，对mysql压力大
 * 2.要求Flink运行的所在机器与MySQL所在的机器时间要同步，不然可能查询不到数据
 *
 * 替代方法（canal 或 Flink CDC）
 */
public class MySQLSource extends RichSourceFunction<GiftBean> {

    private Connection connection;
    private boolean flag = true;
    private long updateTime = 0;

    @Override
    public void open(Configuration parameters) throws Exception {
        //创建JDBC连接
        connection = DriverManager.getConnection("jdbc:mysql://node-3.51doit.cn:3306/aaa?characterEncoding=utf-8", "root", "123456");
    }

    @Override
    public void run(SourceContext<GiftBean> ctx) throws Exception {
        //第一次，把全部的维度数据都读出来，以后只读变化的数据
        PreparedStatement preparedStatement = connection.prepareStatement("SELECT id, name, points, deleted FROM tb_live_gift WHERE updateTime > ?");

        while (flag) {
            //监听MySQL中的变化数据
            preparedStatement.setDate(1, new Date(updateTime));
            ResultSet rs = preparedStatement.executeQuery();
            while (rs.next()) {
                int id = rs.getInt(1);
                String name = rs.getString(2);
                double points = rs.getDouble(3);
                int delete = rs.getInt(4);
                GiftBean bean = GiftBean.of(id, name, points, delete);
                ctx.collect(bean);
            }
            updateTime = System.currentTimeMillis();
            System.out.println(updateTime);
            rs.close();
            //休息
            Thread.sleep(2000);
        }

    }

    @Override
    public void cancel() {
        flag = false;
    }

    @Override
    public void close() throws Exception {
        //关闭连接
        connection.close();
    }
}
